package org.ray.pattern.jobmanager.batch;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

public abstract class AbstractBatchJobManagerImpl<T, Y> implements BatchJobManager<T, Y>
{
	private Map<T, BatchJob<T, Y>> submittedJobMap = new ConcurrentHashMap<T, BatchJob<T, Y>>();
	private Map<T, List<SingleJob<T, Y>>> cachedJobMap = new ConcurrentHashMap<T, List<SingleJob<T, Y>>>();
	private Map<T, FeedbackTask> taskMap = new ConcurrentHashMap<T, FeedbackTask>();
	
	public static final long FEEDBACK_TIME_INTERVAL = 300000L;
	private long feedbackInterval = FEEDBACK_TIME_INTERVAL;

	public void submitBatchJob(BatchJob<T, Y> job)
	{
		// TODO Auto-generated method stub
		submittedJobMap.put(job.getBatchJobKey(), job);
		cachedJobMap.put(job.getBatchJobKey(), new LinkedList<SingleJob<T, Y>>());

		Timer timer = new Timer();
		FeedbackTask task = new FeedbackTask(job.getBatchJobKey(), timer);
		taskMap.put(job.getBatchJobKey(), task);
		timer.schedule(task, feedbackInterval, feedbackInterval);
	}


	public void cancelSingleJob(T batchJobkey, Y singleJobKey)
	{
		// TODO Auto-generated method stub
		BatchJob<T, Y> batchJob = submittedJobMap.get(batchJobkey);
		if (batchJob == null)
		{
			System.out.println("Unregistered Batch Job: " + batchJobkey);
			return;
		}

		synchronized (batchJob)
		{
			batchJob.removeSingleJob(singleJobKey);
		}
	}


	public void reportSingleJobFinished(T batchJobkey, SingleJob<T, Y> singleJob)
	{
		// TODO Auto-generated method stub
		Y singleJobKey = singleJob.getKey();
		if (singleJobKey == null)
		{
			return;
		}
		BatchJob<T, Y> batchJob = submittedJobMap.get(batchJobkey);
		if (batchJob == null)
		{
			System.out.println("Unregistered Batch Job: " + batchJobkey);
			return;
		}

		boolean isInstant = false;
		synchronized (batchJob)
		{
			// Remove single job from submitted store
			batchJob.removeSingleJob(singleJobKey);
			isInstant = batchJob.isEmpty();
			
			// Add the single job to cached store
			List<SingleJob<T, Y>> cachedList = cachedJobMap.get(batchJobkey);
			synchronized (cachedList)
			{
				cachedList.add(singleJob);
			}
			
		}
		
		if (isInstant)
		{
			//When submitted task all finished, run feedback instantly 
			FeedbackTask task = taskMap.get(batchJobkey);
			task.cancelTimer();
			task.run();
		}
		
	}
	
	public long getFeedbackInterval()
	{
		return this.feedbackInterval;
	}

	public void setFeedbackInterval(long lFeedbackInterval)
	{
		this.feedbackInterval = lFeedbackInterval;
	}
	
	protected abstract void feedbackJob(BatchJob<T, Y> batchJob, int unfinishedJobCount, List<SingleJob<T, Y>> finishedJobList);


	public boolean isRegisteredBatchJob(T batchJobkey)
	{
		// TODO Auto-generated method stub
		return (batchJobkey != null) && (submittedJobMap.containsKey(batchJobkey));
	}
	
	class FeedbackTask extends TimerTask
	{
		private Timer myTimer;
		T myBatchJobKey;

		FeedbackTask(T batchJobKey, Timer timer) 
		{
			myTimer = timer;
			myBatchJobKey = batchJobKey;
		}
		
		public void cancelTimer()
		{
			myTimer.cancel();
		}
		
		public void resetTimer(Timer timer)
		{
			myTimer = timer;
		}

		public void run()
		{
			// TODO Auto-generated method stub
			BatchJob<T, Y> batchJob = submittedJobMap.get(myBatchJobKey);
			if (batchJob == null)
				return;
			
			// Move the List to new List and release it
			List<SingleJob<T, Y>> newList = null;
			int unfinishedJobSize = 0;

			synchronized (batchJob)
			{
				List<SingleJob<T, Y>> cachedList = cachedJobMap.get(myBatchJobKey);
				synchronized (cachedList) 
				{
					if (!cachedList.isEmpty() ) 
					{
						newList = new LinkedList<SingleJob<T, Y>>();
						newList.addAll(cachedList);
						cachedList.clear(); //clear the cache
					}
				}

				// When there's no feedback left for the submit, remove the key.
				if (batchJob.isEmpty()) 
				{
					taskMap.remove(myBatchJobKey);
					cachedJobMap.remove(myBatchJobKey);
					submittedJobMap.remove(myBatchJobKey);
				}
				else
				{
					unfinishedJobSize = batchJob.getJobCount();
				}
				
			}

			// do feedback work
			feedbackJob(batchJob, unfinishedJobSize, newList);
		}
	}
}
